package com.flink.stream;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @ClassName flink_wc_socket
 * @Description TODO
 * @Author 长风清留扬
 * @Date 2024/8/8 22:47
 * @Version 1.0
 */
public class FlinkStreamNoLimit {
    public static void main(String[] args) throws Exception {
        //创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //读取数据，从socket中读取数据
        DataStreamSource<String> socket_DS = env.socketTextStream("localhost", 9999);

        //使用lambda表达式来实现
        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = socket_DS.flatMap(
                        (String value, Collector<Tuple2<String, Integer>> out) -> {
                            //拆分
                            String[] words = value.split(" ");
                            for (String word : words) {
                                Tuple2<String, Integer> Tuple2_of = Tuple2.of(word, 1);
                                out.collect(Tuple2_of);
                            }
                        }
                )
                .returns(Types.TUPLE(Types.STRING,Types.INT))
                .keyBy(value -> value.f0)
                .sum(1);

        //输出结果
        sum.print();
        //执行流处理
        env.execute();
    }
}